package com.navinfo.opentsp.platform.computing.serializer;

import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import com.navinfo.opentsp.platform.location.protocol.common.LCStatusType;
import com.navinfo.opentsp.platform.location.protocol.common.LCVehicleStatusData.VehicleStatusData;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.sink.hbase.AsyncHbaseEventSerializer;
import org.apache.hadoop.hbase.util.Bytes;
import org.hbase.async.AtomicIncrementRequest;
import org.hbase.async.PutRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.*;

/**
 * 位置信息LocationData异步序列化
 */
public class AsyncHbaseLocationDataSerializer implements AsyncHbaseEventSerializer {
    private static final Logger log = LoggerFactory.getLogger(AsyncHbaseLocationDataSerializer.class);
    private byte[] table;
    private byte[] colFam = Bytes.toBytes("d");
    private Event currentEvent;

    @Override
    public void initialize(byte[] table, byte[] colFam) {
        this.table = table;
        this.colFam = colFam;
    }

    @Override
    public void setEvent(Event event) {
        this.currentEvent = event;
    }

    @Override
    public void configure(Context context) {
    }
    @Override
    public void configure(ComponentConfiguration componentConfiguration) {
    }
    @Override
    public void cleanUp() {    }

    @Override
    public List<PutRequest> getActions() {
        List<PutRequest> actions = new ArrayList();
        try {
            LCLocationData.LocationData locationData = LCLocationData.LocationData.parseFrom(currentEvent.getBody());

            //rowkey:reverse(tId)|3000000000L-gpsTime
            //reverse(tId)实现rowkey的散列，3000000000L-gpsTime实现逆序最新记录在头部
            String rowKey = new StringBuilder(Long.valueOf(locationData.getTerminalId()).toString()).reverse() + "|" + (3000000000L - locationData.getGpsDate());

            //添加PutRequest
            actions.add(preparedata(rowKey , locationData));
        } catch (Exception e) {
            log.error("LocationData异步序列化失败：" + e.getMessage(), e);
        }

        return actions;
    }

    /**
     * 处理数据
     * @param rowKey
     * @param locationData
     * @throws Exception
     */
    private PutRequest preparedata(String rowKey, LCLocationData.LocationData locationData) throws Exception {
        byte[][] qualifiers = new byte[18][];
        byte[][] values = new byte[18][];

        qualifiers[0] = Bytes.toBytes("alarm");
        qualifiers[1] = Bytes.toBytes("status");
        qualifiers[2] = Bytes.toBytes("longitude");
        qualifiers[3] = Bytes.toBytes("latitude");
        qualifiers[4] = Bytes.toBytes("originalLng");
        qualifiers[5] = Bytes.toBytes("originalLat");
        qualifiers[6] = Bytes.toBytes("height");
        qualifiers[7] = Bytes.toBytes("speed");
        qualifiers[8] = Bytes.toBytes("direction");
        qualifiers[9] = Bytes.toBytes("gpsDate");
        qualifiers[10] = Bytes.toBytes("mileage");
        qualifiers[11] = Bytes.toBytes("oil");
        qualifiers[12] = Bytes.toBytes("recorderSpeed");
        qualifiers[13] = Bytes.toBytes("standardMileage");
        qualifiers[14] = Bytes.toBytes("standardFuelCon");
        qualifiers[15] = Bytes.toBytes("originalPB");
        qualifiers[16] = Bytes.toBytes("tempOilConsumption");
        qualifiers[17] = Bytes.toBytes("lngTotalRemainingGas");

        values[0] = Bytes.toBytes(String.valueOf(locationData.getAlarm()));
        values[1] = Bytes.toBytes(String.valueOf(locationData.getStatus()));
        values[2] = Bytes.toBytes(String.valueOf(locationData.getLongitude()));
        values[3] = Bytes.toBytes(String.valueOf(locationData.getLatitude()));
        values[4] = Bytes.toBytes(String.valueOf(locationData.getOriginalLng()));
        values[5] = Bytes.toBytes(String.valueOf(locationData.getOriginalLat()));
        values[6] = Bytes.toBytes(String.valueOf(locationData.getHeight()));
        values[7] = Bytes.toBytes(String.valueOf(locationData.getSpeed()));
        values[8] = Bytes.toBytes(String.valueOf(locationData.getDirection()));
        values[9] = Bytes.toBytes(String.valueOf(locationData.getGpsDate()));
        values[10] = Bytes.toBytes(String.valueOf(locationData.getMileage()));
        values[11] = Bytes.toBytes(String.valueOf(locationData.getOil()));
        values[12] = Bytes.toBytes(String.valueOf(locationData.getRecorderSpeed()));
        values[13] = Bytes.toBytes(String.valueOf(locationData.getStandardMileage()));
        values[14] = Bytes.toBytes(String.valueOf(locationData.getStandardFuelCon()));
        values[15] = locationData.toByteArray();
        Double tempOilConsumption = 0d;
        try{
            List<VehicleStatusData> list = locationData.getStatusAddition().getStatusList();
            for(VehicleStatusData vehicleStatusData : list){
                if(vehicleStatusData.getTypes() == LCStatusType.StatusType.realTimeOilConsumption){
                    tempOilConsumption = BigDecimal.valueOf(vehicleStatusData.getStatusValue()).divide(BigDecimal.valueOf(100), 2, BigDecimal.ROUND_HALF_UP).doubleValue();
                }
            }
        } catch(Exception e){
            tempOilConsumption = 0d;
        }

        values[16] = Bytes.toBytes(String.valueOf(tempOilConsumption));
        values[17] = Bytes.toBytes(String.valueOf(locationData.getLngTotalRemainingGas()));
        return new PutRequest(table, Bytes.toBytes(rowKey), colFam, qualifiers, values);
    }

    @Override
    public List<AtomicIncrementRequest> getIncrements() {
        return new ArrayList<AtomicIncrementRequest>();
    }

}